package com.bstek.bdf4.mq.service;

import cn.hutool.json.JSONUtil;
import com.bstek.bdf2.core.mq.publisher.BstekSendMessage;
import com.bstek.bdf2.core.mq.publisher.RabbitMQRequestEntity;
import com.bstek.bdf4.mq.core.RabbitMQProPerties;
import com.bstek.bdf4.mq.core.RabbitProducerAckConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;

/**
 * @Title: BstekSendMessageImpl
 * @Package: com.bstek.bdf4.mq.service
 * @Description:
 * @Author: zxq
 * @Date: 2024/2/5 - 14:39
 */

@Primary
@Service
@Slf4j
public class BstekSendMessageImpl implements BstekSendMessage {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private RabbitMQProPerties rabbitMQProPerties;

    @Override
    public void send(RabbitMQRequestEntity requestEntity) {
        log.info("----------开始发送消息,MessageInfo:{}",requestEntity);
//        rabbitTemplate.convertAndSend(messageInfo.getRoutingKey(),messageInfo.getMessage());
        rabbitTemplate.convertAndSend(requestEntity.getExchange(),requestEntity.getRoutingKey(),requestEntity.getData());

    }

    @Override
    public void sendMQ(RabbitMQRequestEntity requestEntity) {
        CorrelationData correlationData = new CorrelationData();
        correlationData.setReturnedMessage(new Message(JSONUtil.toJsonStr(requestEntity).getBytes(),  new MessageProperties()));
        String exchange = requestEntity.getExchange();
        String routingKey = requestEntity.getRoutingKey();
        if (StringUtils.isEmpty(exchange)||StringUtils.isEmpty(routingKey)){
            exchange=rabbitMQProPerties.getExchange();
            routingKey=rabbitMQProPerties.getRoutingKey();
         }
        Assert.hasText(exchange,"exchange is null");
        Assert.hasText(routingKey,"routingKey is null");
        rabbitTemplate.convertAndSend(exchange, routingKey, requestEntity, RabbitProducerAckConfig.messagePostProcessor, correlationData);
    }

    public void sendByCallBack(RabbitMQRequestEntity messageInfo) {
        log.info("----------开始发送消息,MessageInfo:{}",messageInfo);
        // 2.全局唯一的消息ID，需要封装到CorrelationData中
        CorrelationData correlationData = new CorrelationData(messageInfo.getBusinessId());
        // 3.添加callback
        correlationData.getFuture().addCallback(
                result -> {
                    if(result.isAck()){
                        // 3.1.ack，消息成功
                        log.debug("消息发送成功, ID:{}", correlationData.getId());
                    }else{
                        // 3.2.nack，消息失败
                        log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
                    }
                },
                ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
        );
         rabbitTemplate.convertAndSend(messageInfo.getExchange(),messageInfo.getRoutingKey(),messageInfo.getData(),correlationData);

    }

    public void sendTopicExchange(RabbitMQRequestEntity messageInfo){
        String exchangeName = messageInfo.getExchange();
        Assert.hasText(exchangeName,"交换机不能为空！");
        String topic = messageInfo.getRoutingKey();
        Assert.hasText(topic,"topic不能为空！");
        Object message = messageInfo.getData();
        sendTopicExchange(exchangeName,topic,message);

    }

    /**
     * 根据交换机和topic( 通配符)发送消息
     * @param exchangeName 交换机
     * @param topic 通配符
     * @param message 消息
     */
     public void sendTopicExchange(String exchangeName,String topic,Object message){
         rabbitTemplate.convertAndSend(exchangeName,topic,message);
     }

}
